package com.whz.bus.consumer;

import com.whz.bus.util.StructParser;
import io.debezium.engine.DebeziumEngine;
import io.debezium.engine.RecordChangeEvent;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.connect.source.SourceRecord;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.lang.NonNull;
import org.springframework.util.CollectionUtils;

import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;

/**
 * 数据库变化时间的默认消费者,最终处理的数据会发出{@link DataBaseChangeRecordEvent}事件,请自行监听这个事件获取数据库变化数据
 * <p>只监听数据的变化,不监听结构的schema和table变化</p>
 */
@Slf4j
public class DefaultChangeRecordConsumer implements ApplicationContextAware, DebeziumEngine.ChangeConsumer<RecordChangeEvent<SourceRecord>> {
    private ApplicationContext applicationContext;

    @Override
    public void setApplicationContext(@NonNull ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    @Override
    public void handleBatch(List<RecordChangeEvent<SourceRecord>> records, DebeziumEngine.RecordCommitter<RecordChangeEvent<SourceRecord>> committer) throws InterruptedException {

        if (CollectionUtils.isEmpty(records)) {
            committer.markBatchFinished();
            return;
        }

        log.info("receive source records :{} rows", records.size());

        List<RelationalChangeData> changRows = records.stream()
                .map(StructParser::buildFromDebeziumRecord)
                .filter(Objects::nonNull)
                .collect(Collectors.toList());

        if (!CollectionUtils.isEmpty(changRows)) {
            // 发布行变化事件
            try {
                applicationContext.publishEvent(new DataBaseChangeRecordEvent(this, changRows));
            } catch (Throwable throwable) {
                /*
                 *  消费异常,只是简单log下,还是会提交offset.这种处理不是最好的,应该按需处理.
                 */
                log.error("consumer error.", throwable);
            }
        }

        // 必须单个process 后才能 batch finished
        records.forEach(recordRecordChangeEvent -> {
            try {
                committer.markProcessed(recordRecordChangeEvent);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        // 直接确认提交,注意:在用户处理数据异常时,可能导致数据丢失
        committer.markBatchFinished();
        log.debug("commit batch records.");
    }
}